package com.northpool.operator.transfer;

import com.northpool.operator.annotation.InputParam;
import com.northpool.operator.annotation.InputParams;
import com.northpool.operator.annotation.OperatorInfo;
import com.northpool.operator.annotation.ParamEnum;
import com.northpool.operator.annotation.ParamEnums;
import com.northpool.operator.common.database.DatabaseUtils;
import com.northpool.operator.common.stpclient.StpClientOperatorBase;
import com.northpool.operator.param.ParamTool;
import com.northpool.operator.type.TypeEnum;
import com.northpool.resources.Constants;
import com.northpool.resources.datasource.IDataSource;
import com.northpool.stpclient.bean.BaseResponse;
import com.northpool.stpclient.bean.datatransfer.TableCopyBuildPlan;
import com.northpool.stpclient.bean.datatransfer.TbaleCopyModeEnum;
import com.northpool.stpclient.modules.geoprocess.datatransfer.IDataTransferOperator;


@OperatorInfo(
        ref = "Pg2PgOperator",
        name = "Postgres转Postgres",
        description = "",
        enums = {
                @ParamEnums(ref = "mode", value = {
                        @ParamEnum(ref = "REPLACE", name = "删除重建"),
                        @ParamEnum(ref = "MERGE", name = "合并"),
                        @ParamEnum(ref = "APPENDER", name = "追加"),
                })
        }
)
public class Pg2PgOperator extends StpClientOperatorBase {
    String originDataSourceId;
    String originTableName;

    String targetDataSourceId;
    String targetTableName;

    String mode;

    @InputParams({
            @InputParam(ref = "originDataSourceId", name = "来源数据源ID", desc = "来源数据源ID", type = TypeEnum.DATASOURCE, required = true),
            @InputParam(ref = "originTableName", name = "来源表名称", desc = "来源表名称", type = TypeEnum.STRING, required = true),
            @InputParam(ref = "targetDataSourceId", name = "目标数据源ID", desc = "目标数据源ID", type = TypeEnum.DATASOURCE, required = true),
            @InputParam(ref = "targetTableName", name = "目标表名称", desc = "目标表名称", type = TypeEnum.STRING, required = true),
            @InputParam(ref = "mode", name = "传输类型", desc = "传输类型", type = TypeEnum.STRING, value = "REPLACE", enumRef = "mode"),
    })
    public Pg2PgOperator() {
    }

    @Override
    public void execute(ParamTool params) throws Exception {
        originDataSourceId = params.getStr("originDataSourceId");
        originTableName = params.getStr("originTableName");
        targetDataSourceId = params.getStr("targetDataSourceId");
        targetTableName = params.getStr("targetTableName");
        mode = params.getStr("mode");

        IDataTransferOperator transferOperator = this.client.dataTransferOperator();

        String originDataSourceJson = DatabaseUtils.getDataSourceJson(this.originDataSourceId);
        IDataSource originDataSource = IDataSource.fromJSON(originDataSourceJson);

        if (originDataSource == null || !originDataSource.getDataSourceType().equals(Constants.DATA_SOURCE_TYPE.postgreSQL)) {
            throw new RuntimeException(originDataSourceId + ": 数据源未找到或类型不是 postgreSQL");
        }
//        ITable table = originDataSource.getTable(this.originTableName);
//        Integer srid = table.spatialInfo().getSrid();

        String targetDataSourceJson = DatabaseUtils.getDataSourceJson(this.targetDataSourceId);
        IDataSource targetDataSource = IDataSource.fromJSON(targetDataSourceJson);

        TableCopyBuildPlan plan = TableCopyBuildPlan.TableCopyParamsBuilder.create()
//                .addSrid(0)
                // 设置来源数据源
                .addOriginDatasource(originDataSource)
                // 设置来源表
                .addOriginTableName(this.originTableName)
                // 设置目标数据源
                .addTargetDatasource(targetDataSource)
                // 设置目标表名
                .addTargetTableName(this.targetTableName)
                .addMode(TbaleCopyModeEnum.valueOf(mode))
                .build();
        BaseResponse response = transferOperator.startTableCopy(plan);
        String requestId = response.getRequestId();
        transferOperator.onStart(requestId, () -> {
            log("开始数据传输");
        });
        transferOperator.onError(requestId, () -> {
            error(transferOperator.log(requestId));
            this.failedExit();
        });
        transferOperator.onSuccess(requestId, this::successExit);
        this.waiting(24);
    }
}
